package com.ailk.bigdata.etl.realstream.server.service.impl;

import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.PropertyResourceBundle;
import java.util.ResourceBundle;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.constant.Constant;
import com.ailk.bigdata.etl.common.server.constant.InitParamUtil;
import com.ailk.bigdata.etl.realstream.server.constant.ProtocalSdtpConstant;
import com.ailk.bigdata.etl.realstream.server.dao.RealStreamUnitDao;
import com.ailk.bigdata.etl.realstream.server.dao.impl.RealStreamUnitDaoPGImpl;
import com.ailk.bigdata.etl.realstream.server.disruptor.PassivSendHandler;
import com.ailk.bigdata.etl.realstream.server.disruptor.StreamExceptionHandler;
import com.ailk.bigdata.etl.realstream.server.model.ColumnTypeEnum;
import com.ailk.bigdata.etl.realstream.server.model.KafkaInfo;
import com.ailk.bigdata.etl.realstream.server.model.KafkaProduceException;
import com.ailk.bigdata.etl.realstream.server.model.ProtocolType;
import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamFormat;
import com.ailk.bigdata.etl.realstream.server.model.RealStreamUnit;
import com.ailk.bigdata.etl.realstream.server.model.ValueEvent;
import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.ailk.bigdata.etl.realstream.server.protocal.aipr.service.impl.AiprDataAcceptServiceImpl;
import com.ailk.bigdata.etl.realstream.server.protocal.realmq.KafkaDataAcceptServiceImpl;
import com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.impl.LTEDataAcceptServiceImpl;
import com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.impl.SdtpDataAcceptServiceImpl;
import com.ailk.bigdata.etl.realstream.server.service.DataAcceptService;
import com.ailk.bigdata.etl.realstream.server.service.DataDistributeService;
import com.ailk.bigdata.etl.realstream.server.service.FormatHandler;
import com.ailk.bigdata.etl.realstream.server.service.RollingFileService;
import com.ailk.bigdata.etl.realstream.server.thread.PassivSendListenThread;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * 
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-11-3]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class DataDistributeServiceImpl implements DataDistributeService
{
	private Logger logger = LoggerFactory.getLogger(DataDistributeServiceImpl.class);
	private static ConcurrentHashMap<Long, PassivSendListenThread> passivSendListenThs = new ConcurrentHashMap<Long, PassivSendListenThread>();

	public String getAcceptProtocal()
	{
		String acceptProtocal = null;
		try
		{
			String configLocation = "../conf/config.properties";
			BufferedInputStream inputStream = new BufferedInputStream(new FileInputStream(configLocation));
			ResourceBundle bundle = new PropertyResourceBundle(inputStream);
		    acceptProtocal = bundle.getString("realstream_accept_protocal");
		    
			}catch (Exception e) {
				// TODO: handle exception
			}
		return acceptProtocal;
	}
	
	/**
	 * 
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#initDataAcceptThread(com.ailk.bigdata.etl.realstream.server.monitor.EventCounter)
	 */
	@SuppressWarnings("unchecked")
	public void initDataAcceptThread(EventCounter eventCount)
	{
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();
		List<RealStreamUnit> allStreams = streamUnitDao.getAllStreamUnit();
		boolean isAiprStart = false;
		boolean isSdtpCSStart = false;
		boolean isSdtpLTEStart = false;
		//String acceptProtocal = InitParamUtil.getRealStreamAcceptProt();
		String acceptProtocal = getAcceptProtocal();
		List<Integer> protocals = new ArrayList<Integer>();
		if (StringUtils.isNotEmpty(acceptProtocal))
		{
			String[] protocalStr = StringUtils.split(acceptProtocal, ",");
			for (String protocal : protocalStr)
			{
				protocals.add(Integer.parseInt(protocal));
			}
		}
		logger.info("该客户端处理类型[" + StringUtils.join(protocals) + "]");
		for (RealStreamUnit perStream : allStreams)
		{
			logger.info("接口编码[{}],协议类型[{}]启动数据接收服务...........................", perStream.getOnlyUnitCode(),
					ProtocolType.getFixedType(perStream.getProtocolType()));
			if (protocals.contains(perStream.getProtocolType()) && perStream.getProtocolType() == ProtocolType.AIPR.getType() && !isAiprStart)
			{
				// AIPR 协议
				DataAcceptService acceptService = new AiprDataAcceptServiceImpl();
				acceptService.initRealStreamAcceptService(eventCount);
				isAiprStart = true;
				logger.info("A口协议启动成功.........");
			}
			else if (protocals.contains(perStream.getProtocolType()) && perStream.getProtocolType() == ProtocolType.SDTP_CS.getType()
					&& !isSdtpCSStart)
			{
				// SDTP协议
				DataAcceptService acceptService = new SdtpDataAcceptServiceImpl();
				acceptService.initRealStreamAcceptService(eventCount);
				isSdtpCSStart = true;
				logger.info("SDTP口协议启动成功.........");
			}
			else if (protocals.contains(perStream.getProtocolType()) && perStream.getProtocolType() == ProtocolType.SDTP_LTE.getType()
					&& !isSdtpLTEStart)
			{
				// SDTP_LTE协议
				DataAcceptService acceptService = new LTEDataAcceptServiceImpl();
				acceptService.initRealStreamAcceptService(eventCount);
				isSdtpLTEStart = true;
				logger.info("SDTP_LTE口协议启动成功.........");
			}
			else if (protocals.contains(perStream.getProtocolType()) && perStream.getProtocolType() == ProtocolType.KAKFA.getType())
			{
				// kafka 消息接收服务，每个接口单元启动一个
				DataAcceptService acceptService = new KafkaDataAcceptServiceImpl();
				acceptService.initRealStreamAcceptService(eventCount, perStream);
				logger.info("kafka口协议启动成功.........");
			}
		}
	}

	/**
	 * 
	 * @throws KafkaProduceException
	 * @description 初始化数据分发线程(使用 Disruptor 控制)
	 */
	public void initDataDistributeTh(EventCounter eventCounter) throws KafkaProduceException
	{
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();

		KafkaInfo kafkaInfo = null;
		List<RealStreamUnit> streamUnitK = streamUnitDao.getALLKafkaSendUnit();
		if (CollectionUtils.isNotEmpty(streamUnitK))
		{
			for (RealStreamUnit streamUnit : streamUnitK)
			{
				if (StringUtils.isEmpty(streamUnit.getUnitName()))
				{
					logger.warn("单元接口[{}]接口名称不能为空，发送kafka消息服务启动失败", streamUnit.getOnlyUnitCode());
					continue;
				}
				else
				{
					kafkaInfo = streamUnit.getKafkaInfo();
					break;
				}
			}

		}

		// 被动模式处理
		List<RealStreamUnit> allUnits = streamUnitDao.getAllPassivUnit();
		if (CollectionUtils.isNotEmpty(allUnits))
		{

			int num = 60;
			PassivSendHandler[] passiveSendHandler = new PassivSendHandler[num];
			RingBuffer<ValueEvent> buffer = RingBuffer.create(ProducerType.MULTI, ValueEvent.EVENT_FACTORY, 8 << 12, new BlockingWaitStrategy());
			for (int i = 0; i < num; i++)
			{
				passiveSendHandler[i] = new PassivSendHandler(eventCounter);
				passiveSendHandler[i].setKafkaInfo(kafkaInfo);
			}
			WorkerPool<ValueEvent> workerPool = new WorkerPool<ValueEvent>(buffer, buffer.newBarrier(), new StreamExceptionHandler(),
					passiveSendHandler);
			Sequence[] sequences = workerPool.getWorkerSequences();
			buffer.addGatingSequences(sequences);
			workerPool.start(Executors.newFixedThreadPool(num));
			logger.info("数据处理线程池启动成功      处理线程个数[" + num + "]");
			for (RealStreamUnit streamUnit : allUnits)
			{
				RealStreamTool.DISPATCH_CAHCE.put(streamUnit.getOnlyUnitCode(), RealDataDispatcher.addDispatcherHand(streamUnit));
				RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.put(Integer.valueOf(streamUnit.getOnlyUnitCode().toString()), buffer);
			}

		}
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#startInitiativeSendThread()
	 */
	@Override
	public void startInitiativeSendThread()
	{
	}

	@SuppressWarnings("unchecked")
	@Override
	public void startBackToFilesystemThread()
	{
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();
		// 被动模式处理
		List<RealStreamUnit> allUnits = streamUnitDao.getAllPassivUnit();
		if (CollectionUtils.isNotEmpty(allUnits))
		{
			RollingFileService rollingileService = new RollingFileServiceImpl();

			ExecutorService executorService = null;

			//String acceptProtocal = InitParamUtil.getRealStreamAcceptProt();
			String acceptProtocal = getAcceptProtocal();
			List<Integer> protocals = new ArrayList<Integer>();
			if (StringUtils.isNotEmpty(acceptProtocal))
			{
				String[] protocalStr = StringUtils.split(acceptProtocal, ",");
				for (String protocal : protocalStr)
				{
					protocals.add(Integer.parseInt(protocal));
				}
			}
			logger.info("该客户端处理类型[" + StringUtils.join(protocals) + "]");

			for (RealStreamUnit streamUnit : allUnits)
			{
				long onlyUnitCode = streamUnit.getOnlyUnitCode();
				// 落地文件
				if (protocals.contains(streamUnit.getProtocolType())
						&& ProtocalSdtpConstant.REAL_STREAM_IF_CREATE_FILE == streamUnit.getIfCreateFile())
				{
					try
					{
						if (null == executorService)
						{
							executorService = Executors.newCachedThreadPool(new ThreadFactoryBuilder().build());
						}

						RealStreamUnit rStreamUnit = streamUnitDao.getUnitByUnitCode(onlyUnitCode);

						RollingFileWriter fileWriter = rollingileService.getRunnable(RealStreamTool.rollInterval, rStreamUnit.getCreateFileDir(),
								rStreamUnit.getCreateFileProfile());

						if (StringUtils.isNotEmpty(rStreamUnit.getHdfsId()))
						{
							logger.info("上传文件到hdfs");
							rollingileService.uploadHdfs(rStreamUnit);
						}
						else
						{
							// 管理落地文件定时任务，每2小时执行一次
							rollingileService.filePathManager(rStreamUnit, allUnits.size());
						}

						fileWriter.initWriterStreams();
						fileWriter.initRollIntervalTh();
						RealStreamTool.UNITCODE_FILEBACK_BUS_DISRUPTOR.put(onlyUnitCode, fileWriter);
						logger.info("接口onlyUnitCode[" + onlyUnitCode + "]文件落地线程启动成功.");
					}
					catch (Exception e)
					{
						logger.error("配置错误,无法落地生成文件,onlyUnitCode[" + onlyUnitCode + "],请检查相关配置", e);
						System.exit(-1);
					}

				}

			}

		}
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#startPassiveSendListenerTh()
	 */
	@Override
	public void startPassiveSendListenerTh()
	{
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();
		List<RealStreamUnit> streamUnits = streamUnitDao.getAllPassivUnit();
		if (CollectionUtils.isNotEmpty(streamUnits))
		{
			for (RealStreamUnit streamUnit : streamUnits)
			{
				PassivSendListenThread sendListenTh = null;
				if (!passivSendListenThs.containsKey(streamUnit.getOnlyUnitCode()))
				{
					sendListenTh = new PassivSendListenThread(streamUnit.getStreamPassPort(), "thread_" + streamUnit.getOnlyUnitCode());
					passivSendListenThs.put(streamUnit.getOnlyUnitCode(), sendListenTh);
					if (!RealStreamTool.DATAPORT_UNIT_CACHE.containsKey(streamUnit.getStreamPassPort()))
					{
						CopyOnWriteArrayList<RealStreamUnit> realStreamUnitList = new CopyOnWriteArrayList<RealStreamUnit>();
						RealStreamTool.DATAPORT_UNIT_CACHE.put(streamUnit.getStreamPassPort(), realStreamUnitList);
					}
					RealStreamTool.DATAPORT_UNIT_CACHE.get(streamUnit.getStreamPassPort()).add(streamUnit);
				}
				else
				{
					sendListenTh = passivSendListenThs.get(streamUnit.getOnlyUnitCode());
				}
				if (!RealStreamTool.PASSIV_PORT.contains(streamUnit.getStreamPassPort()) && !sendListenTh.isAlive())
				{
					sendListenTh.start();
					RealStreamTool.PASSIV_PORT.add(streamUnit.getStreamPassPort());
				}
				logger.info("被动数据端口[" + streamUnit.getStreamPassPort() + "] 启动");
			}
		}
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#initDataFormatHandler()
	 */
	@Override
	public void initDataFormatHandler()
	{
		// 云南默认被动模式 (先进行处理)
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();
		List<RealStreamUnit> allUnits = streamUnitDao.getAllPassivUnit();
		if (CollectionUtils.isNotEmpty(allUnits))
		{
			for (RealStreamUnit streamUnit : allUnits)
			{
				RealStreamUnitDao unitDao = new RealStreamUnitDaoPGImpl();
				streamUnit = unitDao.getUnitAllByUnitCode(streamUnit.getOnlyUnitCode());
				RealStreamTool.REALS_UNIT_CACHE.put(streamUnit.getOnlyUnitCode(), streamUnit);
				RealStreamFormat acceptformat = streamUnit.getAcceptFormat();
				RealStreamFormat sendformat = streamUnit.getSendFormat();
				String separator = StringUtils.isNotEmpty(sendformat.getDataFormatDefine()) ? sendformat.getDataFormatDefine() : "";
				if (Constant.FILE_ASC_TYPE == acceptformat.getDataFormatType())
				{
					String[] dataFormats = StringUtils.split(acceptformat.getDataFormatDefine(), ";");
					FormatHandler formatHandler = null;
					FormatHandler containerHandler = null;
					for (String dataFormat : dataFormats)
					{
						for (ColumnTypeEnum clmEnum : ColumnTypeEnum.values())
						{
							Pattern patt = Pattern.compile(clmEnum.getPattern());
							Matcher match = patt.matcher(dataFormat);
							if (match.find())
							{
								try
								{
									FormatHandler newFormat = clmEnum.getHandler().newInstance();
									newFormat.setLength(Integer.valueOf(match.group(1)));
									if (match.groupCount() == 2)
									{
										if (clmEnum.equals(ColumnTypeEnum.DATATIME))
										{
											newFormat.setDateFormat(match.group(2));
										}
										else if (clmEnum.equals(ColumnTypeEnum.IP))
										{
											newFormat.setIpVer(Integer.valueOf(match.group(2)));
										}
									}
									newFormat.setSeparator(separator);
									if (formatHandler == null)
									{
										formatHandler = newFormat;
										containerHandler = newFormat;
									}
									else
									{
										containerHandler.setSuccessor(newFormat);
										containerHandler = newFormat;
									}
									break;
								}
								catch (InstantiationException e)
								{
									logger.error("组装责任链异常", e);
								}
								catch (IllegalAccessException e)
								{
									logger.error("组装责任链异常", e);
								}
							}
						}
					}

					RealStreamTool.DATA_FORMAT_HANDLER_MAP.put(streamUnit.getOnlyUnitCode(), formatHandler);

				}
			}
		}
	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#startMessageSendThread()
	 */
	@Override
	public void startMessageSendThread()
	{

	}

	/**
	 * @see com.ailk.bigdata.etl.realstream.server.service.DataDistributeService#initKafkaMessSendTh()
	 */
	public void initKafkaMessSendTh()
	{
		RealStreamUnitDao streamUnitDao = new RealStreamUnitDaoPGImpl();
		List<RealStreamUnit> streamUnits = streamUnitDao.getALLKafkaSendUnit();
		if (CollectionUtils.isNotEmpty(streamUnits))
		{
			for (RealStreamUnit streamUnit : streamUnits)
			{
				if (StringUtils.isEmpty(streamUnit.getUnitName()))
				{
					logger.warn("单元接口[{}]接口名称不能为空，发送kafka消息服务启动失败", streamUnit.getOnlyUnitCode());
					continue;
				}
				RealStreamTool.DATA_SEND_KAFKA_CAHCE.add(streamUnit.getOnlyUnitCode());
				logger.info("单元接口[{}]接口名称[{}]，发送kafka消息服务启动", streamUnit.getOnlyUnitCode(), streamUnit.getUnitName());
			}

		}
	}
}